// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"

#include "row_mutator.hpp"

namespace xbase
{
	void row_mutator::on_async_apply()
	{
		m_location_cache.async_get_meta(
			boost::bind(&this_type::on_got_tablet,
			this_ptr(),
			_1, _2),
			m_table_name,
			m_row_name,
			m_tablet_info_invalid,
			m_meta_data.meta_got_time);
	}

	void row_mutator::on_got_tablet(location_cache_value meta, const error_code &ec)
	{
		if(ec)
		{
			LOG(DEBUG, "got tablet failed");
			if((ec == error::tablet_not_exists) ||
				(ec == error::table_not_exists) ||
				(ec == error::row_not_in_range))
			{
				on_retry(ec);
			}
			else
			{
				on_error(ec);
			}
			return;
		}

		m_meta_data = meta;
		m_tablet_info_invalid = false;

		m_session_manager.async_connect(meta.server_addr_p->to_endpoint(),
			boost::bind(&this_type::on_session_connected,
			this_ptr(),
			_1, _2));
	}

	void row_mutator::on_session_connected(session_ptr session, const error_code &ec)
	{
		if(ec)
		{
			LOG(DEBUG, "session connect failed");
			on_retry(ec);
			return;
		}

		// set tablet name
		m_req_ptr->body().m_tablet_name = m_meta_data.tablet_info_p->tablet_name();

		if(! m_tablet_mutator)
			m_tablet_mutator.reset(new_async_module(row_mutator_on_tablet, m_io_service_pool));
		m_tablet_mutator->async_apply(session,
			m_req_ptr,
			boost::bind(&this_type::on_complete,
			this_ptr(),
			_1));
	}

	void row_mutator::on_complete(const error_code &ec)
	{
		if(ec)
		{
			LOG(ERROR, "async apply on tablet mutator failed");
			if((ec == error::tablet_not_exists) ||
				(ec == error::table_not_exists))
			{
				on_retry(ec);
			}
			else
			{
				on_error(ec);
			}
		}
		else
		{
			dispatch(ec);
		}
	}

	void row_mutator::on_retry(const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "error occur in row_mutator, table_name = "
				<< m_table_name << ", mutation row = " << dump_string_char_with_hex(m_row_name)
				<< ", tablet info = "
				<< (m_meta_data.tablet_info_p ? m_meta_data.tablet_info_p->to_string() : "NULL")
				<< ", server address = "
				<< (m_meta_data.server_addr_p ? m_meta_data.server_addr_p->to_string() : "NULL")
				<< ", meta_got_time = " << m_meta_data.meta_got_time
				<< ", tablet_info_invalid = " << std::boolalpha << m_tablet_info_invalid
				<< ", retry_times = " << m_retry_times
				<< ", error = " << ec << " - " << ec.message());

			++m_retry_times;
			m_tablet_info_invalid = true;

			if(! m_timer)
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));
			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::on_async_apply, this_ptr()));
		}
		else
		{
			on_error(ec);
		}
	}

	void row_mutator::on_error(const error_code &ec)
	{
		LOG(ERROR, "error occur in row_mutator, table_name = "
			<< m_table_name << ", mutation row = " << dump_string_char_with_hex(m_row_name)
			<< ", tablet info = "
			<< (m_meta_data.tablet_info_p ? m_meta_data.tablet_info_p->to_string() : "NULL")
			<< ", server address = "
			<< (m_meta_data.server_addr_p ? m_meta_data.server_addr_p->to_string() : "NULL")
			<< ", meta_got_time = " << m_meta_data.meta_got_time
			<< ", tablet_info_invalid = " << std::boolalpha << m_tablet_info_invalid
			<< ", retry_times = " << m_retry_times
			<< ", error = " << ec << " - " << ec.message());

		dispatch(ec);
	}

	void row_mutator::dispatch(const error_code &ec)
	{
		// release resources
		m_req_ptr.reset();
		handler_type tmp = m_handler;
		m_handler = handler_type(); // clear handler
		this->get_upper_level_service().dispatch(boost::bind(tmp, ec));
	}
}
